Data Engineering Path · Airflow
Sensors — Waiting for External Conditions
⏳ Sensors Monitor and Wait
A Sensor is a special type of operator that waits for a specific condition to be met before allowing downstream tasks to execute. Think of sensors as "gatekeepers" in your pipeline.
Sensor Modes
| Mode | Behavior | Worker Usage | Best For |
|---|---|---|---|
| poke | Keeps checking at poke_interval intervals, holds worker slot |
⚠️ Blocks 1 worker slot | Short waits (< 5 minutes) |
| reschedule | Releases worker between checks, re-enters queue at poke_interval |
✅ Frees worker between checks | Medium waits (5-60 minutes) |
| deferrable | Hands off to Triggerer completely | ✅ Zero worker usage | Long waits (hours) |
# Mode comparison — same sensor, different modes
# ❌ Poke mode — blocks a worker for up to 1 hour
wait_poke = S3KeySensor(
task_id="wait_poke",
bucket_name="data",
bucket_key="input.csv",
mode="poke",
poke_interval=60,
timeout=3600,
)
# ✅ Reschedule mode — releases worker between checks
wait_reschedule = S3KeySensor(
task_id="wait_reschedule",
bucket_name="data",
bucket_key="input.csv",
mode="reschedule",
poke_interval=300,
timeout=3600,
)
# ✅✅ Deferrable mode — best option (Airflow 2.2+)
wait_deferrable = S3KeySensor(
task_id="wait_deferrable",
bucket_name="data",
bucket_key="input.csv",
deferrable=True,
timeout=3600,
)
Common Sensors
| Sensor | Provider | What It Waits For |
|---|---|---|
S3KeySensor |
AWS | File exists in S3 bucket |
HttpSensor |
HTTP | API endpoint returns expected response |
FileSensor |
Standard | File exists on local filesystem |
SqlSensor |
Standard | SQL query returns a truthy result |
ExternalTaskSensor |
Standard | A task in another DAG completes |
DateTimeSensor |
Standard | A specific datetime is reached |
BigQueryTableExistenceSensor |
GCP | Table exists in BigQuery |
Practical Example: Data Arrival Pipeline
from airflow.sdk import DAG
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.providers.standard.sensors.http import HttpSensor
from airflow.providers.standard.operators.python import PythonOperator
from datetime import datetime, timedelta
with DAG(
"data_arrival_pipeline",
schedule="@daily",
start_date=datetime(2024, 1, 1),
default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
) as dag:
# 1. Wait for upstream API to be ready
check_api_health = HttpSensor(
task_id="check_api_health",
http_conn_id="data_api",
endpoint="/health",
response_check=lambda response: response.json()["status"] == "healthy",
deferrable=True,
timeout=1800, # Wait up to 30 minutes
)
# 2. Wait for data file to arrive in S3
wait_for_data = S3KeySensor(
task_id="wait_for_sales_data",
bucket_name="raw-data-{{ var.value.environment }}",
bucket_key="sales/{{ ds }}/data.parquet",
aws_conn_id="aws_default",
deferrable=True,
timeout=7200, # Wait up to 2 hours
)
# 3. Process the data once it arrives
process = PythonOperator(
task_id="process_sales_data",
python_callable=run_spark_job,
)
[check_api_health, wait_for_data] >> process
💡 Tip
Always set a
Always set a
timeout on sensors. Without a timeout, a sensor will wait forever, blocking your pipeline. A good practice is to set the timeout to the maximum reasonable wait time for your data source.